/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package com.google.code.bigyang.nio.impl;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import com.google.code.bigyang.base.SessionState;
import com.google.code.bigyang.config.Configuration;
import com.google.code.bigyang.config.ExtendConfiguration;
import com.google.code.bigyang.config.SocketOptionSet;
import com.google.code.bigyang.log.Log;
import com.google.code.bigyang.nio.NioAcceptor;
import com.google.code.bigyang.nio.NioSession;
import com.google.code.bigyang.nio.NioSessionEvent;
import com.google.code.bigyang.util.ThreadUtil;

public class NioTcpAcceptor extends NioAcceptor implements Runnable {

	private ServerSocketChannel serverChannel;
	
	public NioTcpAcceptor(Configuration config) {
		super(config);
	}
	
	@Override
	public void start() {
		try {
			configServerChannel();
			manager.start();
			for(int i=0;i<ExtendConfiguration.getAcceptorNum();i++) {
				ThreadUtil.createThread(this, "acceptor-"+(i+1));
			}
			
		} catch (IOException e) {
			Log.error(e.getMessage(),e);
		}
	}
	
	public NioSession createSessionInstance() {
		return new NioTcpSession();
	}

	public void run() {
		while(true) {
			try {
				SocketChannel channel = serverChannel.accept();
				
				configChannel(channel);
				
				NioSession session = createSession(channel);
				session.handleEvent(SessionState.CONNECTED, null);
				NioSessionEvent event = new NioSessionEvent(session, SessionState.ENABLE_READ);
				manager.registerEvent(event);
			} catch (IOException e) {
				Log.error(e.getMessage(),e);
			}
		}
		
	}

	private void configChannel(SocketChannel channel) throws IOException {		
		channel.configureBlocking(false);
		Socket socket = channel.socket();
		socket.setSoTimeout(config.getSoTimeout());
		
		if(config.getOptions().get(SocketOptionSet.SO_KEEPALIVE) != null) {
			socket.setKeepAlive( (Boolean) SocketOptionSet.SO_KEEPALIVE.type().cast(
				config.getOptions().get(SocketOptionSet.SO_KEEPALIVE)	
			) );
		}
		
		if(config.getOptions().get(SocketOptionSet.SO_LINGER) != null) {
			socket.setSoLinger( true, (Integer) SocketOptionSet.SO_LINGER.type().cast(
				config.getOptions().get(SocketOptionSet.SO_LINGER)
			) );
		}
		
		if(config.getOptions().get(SocketOptionSet.SO_RCVBUF) != null) {
			socket.setReceiveBufferSize((Integer) SocketOptionSet.SO_RCVBUF.type().cast(
					config.getOptions().get(SocketOptionSet.SO_RCVBUF)
			) );
		}
		
		if(config.getOptions().get(SocketOptionSet.SO_SNDBUF) != null) {
			socket.setSendBufferSize( (Integer) SocketOptionSet.SO_SNDBUF.type().cast(
					config.getOptions().get(SocketOptionSet.SO_SNDBUF)
			) );
		}
		
		if(config.getOptions().get(SocketOptionSet.SO_REUSEADDR) != null) {
			socket.setReuseAddress( (Boolean) SocketOptionSet.SO_REUSEADDR.type().cast(
					config.getOptions().get(SocketOptionSet.SO_REUSEADDR)
			) );
		}
		
		if(config.getOptions().get(SocketOptionSet.TCP_NODELAY) != null) {
			socket.setTcpNoDelay( (Boolean) SocketOptionSet.TCP_NODELAY.type().cast(
					config.getOptions().get(SocketOptionSet.TCP_NODELAY)
			) );
		}
		
	}
	
	
	private void configServerChannel() throws IOException {
		serverChannel = ServerSocketChannel.open();
		serverChannel.socket().bind(new InetSocketAddress(config.getAcceptorPort()));
		serverChannel.configureBlocking(true);
	}
	
	
	
	
	
}
